package com.atguigu.edu.realtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.BaseAppV1;
import com.atguigu.edu.realtime.bean.TableProcess;
import com.atguigu.edu.realtime.common.Constant;
import com.atguigu.edu.realtime.util.FlinkSinkUtil;
import com.atguigu.edu.realtime.util.JdbcUtil;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.Arrays;
import java.util.List;

public class DimApp extends BaseAppV1 {
    public static void main(String[] args) {
        new DimApp().init(2001,2,"DimApp", Constant.TOPIC_ODS_DB);
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 对流做操作
        // 1. 对业务数据做过滤 ETL
        SingleOutputStreamOperator<JSONObject> etledStream = etl(stream);
        // 2. 读取配置信息
        SingleOutputStreamOperator<TableProcess> tpStream = readTableProcess(env);
        // 3. 数据流和广播流做connect,同时预先根据配置在hbase中建表
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dataTpStream = connect(etledStream, tpStream);
        // 4. 过滤掉不需要的字段
        SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream = filterNotNeedColumns(dataTpStream);
        // 5. 根据不同的配置信息, 把不同的维度写入到不同的Phoenix的表中
        writeToPhoenix(resultStream);
    }

    private void writeToPhoenix(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> resultStream) {
        resultStream.addSink(FlinkSinkUtil.getPhoenixSink());
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> filterNotNeedColumns(SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> dataTpStream) {
        return dataTpStream
                .map(new MapFunction<Tuple2<JSONObject, TableProcess>, Tuple2<JSONObject, TableProcess>>() {
                    @Override
                    public Tuple2<JSONObject, TableProcess> map(Tuple2<JSONObject, TableProcess> value) throws Exception {
                        JSONObject data = value.f0;
                        List<String> columns = Arrays.asList(value.f1.getSinkColumns().split(","));

                        // data其实是一个map, 从map中删除键值对
                        data.keySet().removeIf(key -> {
                            return !columns.contains(key) && !"op_type".equals(key);
                        });

                        return value;
                    }
                });
    }

    private SingleOutputStreamOperator<Tuple2<JSONObject, TableProcess>> connect(SingleOutputStreamOperator<JSONObject> dataStream, SingleOutputStreamOperator<TableProcess> tpStream) {
        //1.根据配置表中的信息，在phoenix中创建相应维度表(以免广播之后多处建表)
        tpStream.map(new RichMapFunction<TableProcess, TableProcess>() {

            private Connection conn;

            @Override
            public void open(Configuration parameters) throws Exception {
                //获取connection
                conn = JdbcUtil.getPhoenixConnection();
            }

            @Override
            public void close() throws Exception {
                //归还
                JdbcUtil.closeConnection(conn);
            }

            @Override
            public TableProcess map(TableProcess tp) throws Exception {
                // 避免与服务器的长连接, 长时间没有使用, 服务器会自动关闭连接.
                if (conn.isClosed()) {
                    conn = JdbcUtil.getPhoenixConnection();
                }

                //拼接SQL语句
                StringBuilder sql = new StringBuilder();
                sql
                        .append("create table if not exists ")
                        .append(tp.getSinkTable())
                        .append("(")
                        .append(tp.getSinkColumns().replaceAll("[^,]+","$0 varchar"))
                        .append(",constraint pk primary key(")
                        .append(tp.getSinkPk() == null ? "id" : tp.getSinkPk())
                        .append("))")
                        .append(tp.getSinkExtend() == null ? "" : tp.getSinkExtend());

                System.out.println("phoenix 建表语句："+sql);

                PreparedStatement ps = conn.prepareStatement(sql.toString());

                ps.execute();
                ps.close();
                return tp;
            }
        });


        // 1. 把配置流做成广播流
        MapStateDescriptor<String, TableProcess> tpStateDesc = new MapStateDescriptor<>("tpState", String.class, TableProcess.class);
        BroadcastStream<TableProcess> tpBcStream = tpStream.broadcast(tpStateDesc);
        //2.数据就connect广播流
        return dataStream
        //connect
        .connect(tpBcStream)
        //数据处理
        .process(new BroadcastProcessFunction<JSONObject, TableProcess, Tuple2<JSONObject,TableProcess>>() {
            // 处理数据流中的元素
            @Override
            public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<Tuple2<JSONObject, TableProcess>> collector) throws Exception {
                //处理数据流中数据的时候, 从广播状态读取他对应的配置信息
                //根据什么获取到配置信息: 根据mysql中的表名
                ReadOnlyBroadcastState<String, TableProcess> state = ctx.getBroadcastState(tpStateDesc);
                String key = value.getString("table");

                TableProcess tp = state.get(key);
                if (tp != null) {
                    JSONObject data = value.getJSONObject("data");
                    //操作类型 加到json中
                    data.put("op_type",value.getString("type"));

                    collector.collect(Tuple2.of(data,tp));
                }


            }

            @Override
            public void processBroadcastElement(TableProcess value, Context context, Collector<Tuple2<JSONObject, TableProcess>> collector) throws Exception {
                //把配置信息写入到广播部状态
                String key = value.getSourceTable();
                BroadcastState<String, TableProcess> state = context.getBroadcastState(tpStateDesc);
                state.put(key,value);
            }
        });
    }

    private SingleOutputStreamOperator<TableProcess> readTableProcess(StreamExecutionEnvironment env) {
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname(Constant.MYSQL_HOST)
                .port(3306)
                .databaseList("edu_config") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("edu_config.table_process") // set captured table
                .username(Constant.MYSQL_USERNAME)
                .password(Constant.MYSQL_PASSWORD)
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();

        return env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
                .map(json -> {
                    JSONObject obj = JSON.parseObject(json);
                    return obj.getObject("after", TableProcess.class);
                });
    }

    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> stream) {
        return stream
                .filter(json -> {
                    try {
                        JSONObject obj = JSON.parseObject(json.replaceAll("bootstrap-", ""));

                        return "edu".equals(obj.getString("database"))
                                && ("insert".equals(obj.getString("type")) || "update".equals(obj.getString("type")))
                                && obj.getString("data") != null
                                && obj.getString("data").length() > 2;

                    } catch (Exception e) {
                        System.out.println("json 格式有误，已滤除：" + json);
                        return false;
                    }
                })
                .map(JSON::parseObject);
    }
}
